/*
 *  Copyright 2009 Brian Mackay brianwmackay@msn.com
 *   All Rights Reserved. See end of file
 */
package com.bmackay.vonsteuben.job.control;

import com.bmackay.semweb.commons.decorators.jena.ModelLoader;
import com.bmackay.semweb.commons.decorators.jena.graph.query.GraphQuery;
import com.bmackay.vonsteuben.base.conf.ConfigParserImpl;
import com.bmackay.vonsteuben.job.mapred.base.DriverJobConf;
import com.bmackay.vonsteuben.job.mapred.base.JobDrivers;
import com.bmackay.vonsteuben.model.hdfs.HdfsModel;
import com.bmackay.vonsteuben.ontology.VonSteuben;
import com.bmackay.vonsteuben.ontology.triples.VonSteubenTriples;
import com.hp.hpl.jena.graph.Node;
import com.hp.hpl.jena.graph.Triple;
import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.vocabulary.RDF;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

public class StepProccessing {

    HdfsModel hdfsModel = new HdfsModel();

    public void execute(Node nextProcessNode, Model model, JobDrivers jobDrivers) {

        Node type = determineType(model, nextProcessNode);

        if (type.equals(VonSteuben.ResourceURI.asNode())) {

            Node uri = GraphQuery.getObect(model, VonSteubenTriples.hasURI(nextProcessNode));
            //todo: add web links in addition to local files
            Model linkedModel = ModelLoader.loadModelFullPath(uri.getLiteralLexicalForm());
            type = determineType(linkedModel, Node.ANY);
            processResource(type, jobDrivers, Node.ANY, linkedModel);
        } else {
            processResource(type, jobDrivers, nextProcessNode, model);
        }
    }

    private void processResource(Node type, JobDrivers jobDrivers,
            Node nextProcessNode, Model model) {

        if (type.equals(VonSteuben.HDFS.asNode())) {
            executeHDFSModel(model, nextProcessNode);
        }
        if (type.equals(VonSteuben.Job.asNode())) {
            addJobDrivers(jobDrivers, model, nextProcessNode);
        }
    }

    private Node determineType(Model model, Node nextProcessNode) {
        return GraphQuery.getObect(model, Triple.create(nextProcessNode, RDF.type.asNode(), Node.ANY));
    }

    private void addJobDrivers(JobDrivers jobDrivers, Model model, Node nextProcessNode) {
        jobDrivers.add(getDriverJobConf(model, nextProcessNode));
    }

    private void executeHDFSModel(Model model, Node nextProcessNode) {
        hdfsModel.execute(model, nextProcessNode);
    }

    private DriverJobConf getDriverJobConf(Model model, Node n) {

        Node driverURI = GraphQuery.getObect(model, VonSteubenTriples.hasDriver(n));
        DriverJobConf driverJobConf = new DriverJobConf(driverURI.getLiteralLexicalForm());
        setAddedConf(model, n, driverJobConf);
        return driverJobConf;
    }

    private void setAddedConf(Model model, Node n, DriverJobConf driverJobConf) {

        Node addedConf = GraphQuery.getObect(model, VonSteubenTriples.hasAdditionalConfiguration(n));

        if (addedConf != null) {
            ConfigParserImpl ConfigParserImpl = new ConfigParserImpl(addedConf.getLiteralLexicalForm(), false);
            List<Map.Entry<String, String>> list = ConfigParserImpl.getAll();

            if (list != null) {
                for (Iterator<Map.Entry<String, String>> it = list.iterator(); it.hasNext();) {
                    Map.Entry<String, String> entry = it.next();
                    driverJobConf.set(entry.getKey(), entry.getValue());
                }
            }
        }
    }
}
/*
 * Copyright 2009 Brian Mackay brianwmackay@msn.com
 * All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the 'License');
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an 'AS IS' BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */